from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode

#1、创建flink环境
env = StreamExecutionEnvironment.get_execution_environment()

#设置任务的并行度，一个并行的对应一个task

env.set_parallelism(1)

#设置处理模式
#BATCH：批处理模式
#1、输出最终计算结果
#2、内部的计算方式和MR一样，分为map和reduce端进行计算
#3、批处理模式只能用于处理有界流

#STREAMING：流处理模式
#1、输出持续结果，输出中间结果
#2、内部是一条一条计算
#3、流处理模式既能处理有界流也能处理无界流

env.set_runtime_mode(RuntimeExecutionMode.BATCH)

#2、读取数据
#有界流
lines_ds = env.read_text_file('../../data/word.txt')

#3、统计单词的梳理
#一行转换成多行
words_ds = lines_ds.flat_map(lambda line: line.split(','))

#转换成kv格式
kv_ds = words_ds.map(lambda word: (word, 1))

#安装单词分组
key_by_ds = kv_ds.key_by(lambda kv: kv[0])

#统计单词数量
count_ds = key_by_ds.sum(1)
#打印数据
count_ds.print()

#启动flink任务
env.execute()

